package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.dto.UserDTO;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisConstants;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.SimpleRedisLock;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    
    @Resource
    private SeckillVoucherServiceImpl seckillVoucherService;
    @Resource
    private RedisIdWorker redisIdWorker;
    @Resource(name = "stringRedisTemplate")
    private StringRedisTemplate redisTemplate;
    @Resource
    private RedissonClient redissonClient;

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;//判断下单资格lua脚本
    private static final List<String> KEYS;

    static {//加载类时初始化lua脚本
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//配置路径
        SECKILL_SCRIPT.setResultType(Long.class);//设置返回类型
        KEYS = new ArrayList<>();
    }
    //自定义ThreadFactory
    private static class MyThreadFactory implements ThreadFactory {
        private AtomicInteger threadNumber = new AtomicInteger(1);
        private String threadNamePrefix;

        public MyThreadFactory(String threadNamePrefix) {
            this.threadNamePrefix = threadNamePrefix;
        }

        @Override
        public Thread newThread(Runnable runnable) {
            // 创建线程，设置自定义名称
            return new Thread(runnable, threadNamePrefix + threadNumber.getAndIncrement());
        }
    }

//    单个线程的线程池，用于在数据库扣减库存、插入订单
    //private static final ExecutorService SECKILL_ORDER_EXECUTORS = Executors.newSingleThreadExecutor();
    private static final ExecutorService SECKILL_ORDER_EXECUTORS =
        new ThreadPoolExecutor(4,
                        4,
                        0,
                        TimeUnit.SECONDS,
                        new ArrayBlockingQueue<>(8),
                        new MyThreadFactory("c"),
                        new ThreadPoolExecutor.AbortPolicy());

    private static final String STREAM_NAME = "stream-orders";

    private IVoucherOrderService currentProxy;

    private class Task implements Runnable {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName());
            while (true) {
                try {
                    //获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream-orders >
                    List<MapRecord<String, Object, Object>> list = redisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1L).block(Duration.ofSeconds(2L)),
                            StreamOffset.create(STREAM_NAME, ReadOffset.lastConsumed())
                    );
                    //消息获取失败，没有消息，继续下一次循环
                    if (list == null || list.isEmpty()) {
                        continue;
                    }
                    //消息获取成功，创建订单
                    MapRecord<String, Object, Object> orderInfo = list.get(0);//解析订单信息
                    Map<Object, Object> kv = orderInfo.getValue();//获取消息中的键值对
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(kv, new VoucherOrder(), false);//Map对象转为voucherOrder对象
                    currentProxy.executeCreateVoucherOrder(voucherOrder);
                    //ACK确认，告诉消息队列处理完成 SACK stream-orders g1 消息ID
                    redisTemplate.opsForStream().acknowledge(STREAM_NAME, "g1", orderInfo.getId());

                } catch (Exception e) {
                    log.error("订单处理异常", e);
                    while (true) {
                        try {
                            //在pending-list中获取未确认的消息 XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream-orders 0
                            List<MapRecord<String, Object, Object>> list1 = redisTemplate.opsForStream().read(
                                    Consumer.from("g1", "c1"),
                                    StreamReadOptions.empty().count(1L),
                                    StreamOffset.create(STREAM_NAME, ReadOffset.from("0"))
                            );
                            //pending-list中没有消息，结束，继续正常获取消息队列中的订单信息
                            if (list1 == null || list1.isEmpty()) {
                                break;
                            }
                            //pending-list中的消息获取成功，创建订单
                            MapRecord<String, Object, Object> orderInfo1 = list1.get(0);//解析订单信息
                            Map<Object, Object> kv = orderInfo1.getValue();//获取消息中的键值对
                            VoucherOrder voucherOrder1 = BeanUtil.fillBeanWithMap(kv, new VoucherOrder(), false);//Map对象转为voucherOrder对象
                            currentProxy.executeCreateVoucherOrder(voucherOrder1);
                            //ACK确认，告诉消息队列处理完成 SACK stream-orders g1 消息ID
                            redisTemplate.opsForStream().acknowledge(STREAM_NAME, "g1", orderInfo1.getId());
                        } catch (Exception ex) {
                            //处理pending-list中的消息过程中发生异常，继续while循环，处理pending-list
                            log.error("处理pending-list中的消息过程中发生异常", ex);
                        }
                    }
                }
            }
        }
    }

    //在类初始化完成后执行
    @PostConstruct
    private void init() {
        for (int i = 0; i < 1; i++) {
            SECKILL_ORDER_EXECUTORS.submit(new Task());
        }
    }

//    @PreDestroy
//    private void destroy(){
//        SECKILL_ORDER_EXECUTORS.shutdown();
//        try {
//            // 等待线程池关闭，最多等待5分钟
//            if (!SECKILL_ORDER_EXECUTORS.awaitTermination(5, TimeUnit.SECONDS)) {
//                // 如果等待超时，则打印日志
//                log.error("线程池未能在5s内完全关闭");
//            }
//        } catch (InterruptedException e) {
//            // 异常处理
//            e.printStackTrace();
//        }
//    }

    /**
     * 新开线程完成库存减少、创建优惠券订单，无需返回值
     * @param voucherOrder
     * @return
     */
    @Transactional
    public void executeCreateVoucherOrder(VoucherOrder voucherOrder) {
        //扣减库存
        boolean success = seckillVoucherService.update().
                setSql("stock = stock - 1").
                eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).
                update();
        //插入订单
        save(voucherOrder);//写入数据库
    }

    /**
     * 优惠券秒杀
     * 同一个userId，最多下一单
     * @param voucherId
     * @return
     */
    @Override
    public Result seckill(Long voucherId) {
        //获取代理对象
        currentProxy = (IVoucherOrderService) AopContext.currentProxy();

        //获取用户
        Long userId = UserHolder.getUser().getId();//用户id
        long orderId = redisIdWorker.nextId(RedisConstants.ORDER_KEY);//订单id
        //调用seckill.lua 判断下单资格、redis库存预扣减、向消息队列发消息（优惠券id，用户id，订单id）
        Long result = redisTemplate.execute(SECKILL_SCRIPT, KEYS, voucherId.toString(), userId.toString(), String.valueOf(orderId));
        //判断结果是否为0，不为0代表没有下单资格(1库存不足，2重复下单)
        if (result != 0L) {
            return (result == 1L)? Result.fail("库存不足") : Result.fail("重复下单");
        }

//        //为0，有购买资格，将下单信息保存到阻塞队列
//        VoucherOrder voucherOrder = new VoucherOrder();
//        long orderId = redisIdWorker.nextId(RedisConstants.ORDER_KEY);//订单id
//        voucherOrder.setId(orderId);
//        voucherOrder.setUserId(userId);//用户id
//        voucherOrder.setVoucherId(voucherId);//代金券id
//        orderTaskQueue.add(voucherOrder);//将下单信息保存到阻塞队列
        //返回订单id
        return Result.ok(orderId);
    }

    /**
     * 优惠券秒杀 优化前
     * 同一个userId，最多下一单
     * @param voucherId
     * @return
     */
//    @Override
//    public Result seckill(Long voucherId) {
//        //1. 查询代金券是否存在
//        //SeckillVoucher dbSeckillVoucher = seckillVoucherService.query().eq("voucher_id", voucherId).one();
//        SeckillVoucher dbSeckillVoucher = seckillVoucherService.getById(voucherId);
//        if (dbSeckillVoucher == null) {
//            return Result.fail("优惠券不存在！");
//        }
//        //2. 判断秒杀是否开始/结束
//        if (dbSeckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
//            return Result.fail("秒杀尚未开始！");
//        }
//        if (dbSeckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
//            return Result.fail("秒杀已结束！");
//        }
//        //3. 判断库存数量是否充足
//        if (dbSeckillVoucher.getStock() <= 0) {
//            return Result.fail("优惠券已售完。");
//        }
//        /*
//        //在判断一人一单时，要加锁，否则会出现userId相同的多个线程
//        //都查询到userOrderCount=0，从而创建多个订单的问题。
//        //由于是数据库的插入操作，所以不能通过查询数据版本号的方法加乐观锁。
//        //这里采用悲观锁。
//        //锁对象应该是userId，因为只是希望相同userId的线程串行执行该方法
//        //1. 注意保证userId相同时，锁是同一个对象。
//        //2. 注意加锁的范围，应该截至到事务提交(数据库数据改变)完成后，
//        //如果加锁范围截至到事务提交前，则在订单还未插入数据库时，锁
//        //就被释放了，此时再来一个相同userId的线程，查到的userOrderCount
//        //还是0，还是会出现下多单的情况。
//        Long userId = UserHolder.getUser().getId();//用户id
//        //如果在常量池中已经存在相同内容的字符串，intern() 方法将返回常量池中的字符串对象的引用。
//        //如果常量池中不存在相同内容的字符串，intern() 方法则会将该字符串对象包含的字符串添加到常量池中，并返回它的引用。
//        //保证userId相同时，锁是同一个对象。
//        synchronized (userId.toString().intern()) {
//            IVoucherOrderService currentProxy = (IVoucherOrderService) AopContext.currentProxy();
//            return currentProxy.createVoucherOrder(voucherId);//为了保证事务生效，需要获取代理对象，并调用代理对象的方法
//        }
//         */
//        //解决服务器集群时的线程安全问题：synchronized不再适用
//        //创建锁对象 保证userId相同时，用同一个锁(在redis中有相同的key)
//        Long userId = UserHolder.getUser().getId();//用户id
//        //SimpleRedisLock lock = new SimpleRedisLock(redisTemplate, "order:" + userId);
//        RLock lock = redissonClient.getLock("lock:order:" + userId);
//        //尝试获取锁
//        //boolean getLockSuccess = lock.tryLock(1200);
//        boolean getLockSuccess = lock.tryLock();
//        //判断是否获取锁成功
//        if (!getLockSuccess) {
//            //获取锁失败，返回错误
//            return Result.fail("对于同一张优惠券，您最多只能购买一次。");
//        }
//        //业务执行过程中可能出现异常，即使出现异常也要释放锁
//        try {
//            IVoucherOrderService currentProxy = (IVoucherOrderService) AopContext.currentProxy();
//            return currentProxy.createVoucherOrder(voucherId);//为了保证事务生效，需要获取代理对象，并调用代理对象的方法
//        } finally {
//            lock.unlock();
//        }
//    }

    /**
     * 完成库存减少、创建优惠券订单
     * @param voucherId
     * @return
     */
    @Transactional
    public Result createVoucherOrder(Long voucherId) {
        //3.1 根据userId和优惠券id，判断该用户是否已经下过单
        Long userId = UserHolder.getUser().getId();//用户id
        Integer userOrderCount = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        //3.2 一个人最多下一单
        if (userOrderCount > 0) {
            return Result.fail("对于同一张优惠券，您最多只能购买一次。");
        }
        //4. 扣减库存：乐观锁——只有当前库存未被修改过时，才进行更新操作 修改时，限制stock=dbSeckillVoucher.getStock()
        //乐观锁问题：成功率太低
        //改进：更新操作时，限制stock>0
        //（数据库会加锁 一个进程在修改时其他进行不能修改，所以当你查询到时stock不大于0，其他进行已经进行了修改）
        boolean success = seckillVoucherService.update().
                setSql("stock = stock - 1").
                eq("voucher_id", voucherId).gt("stock", 0).
                update();
        if (!success) {
            return Result.fail("优惠券已售完。");
        }
        //5. 创建订单tb_voucher_order
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIdWorker.nextId(RedisConstants.ORDER_KEY);//订单id
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(userId);//用户id
        voucherOrder.setVoucherId(voucherId);//代金券id
        save(voucherOrder);//写入数据库
        //6.返回订单id
        return Result.ok(orderId);
    }
}
